package com.shujia.flink.sink

import java.util

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import redis.clients.jedis.Jedis

object Demo3Redis {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val ds: DataStream[String] = env.readTextFile("flink/data/students.txt")

    ds.addSink(new RedisSink)

    env.execute()
  }

}

class RedisSink extends RichSinkFunction[String] {

  var jedis: Jedis = _

  override def open(parameters: Configuration): Unit = {

    //创建连接
    jedis = new Jedis("master", 6379)

    jedis.auth("123456")
  }

  override def close(): Unit = {
    jedis.close()
  }

  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {


    val split: Array[String] = value.split(",")


    val id: String = split(0)

    val map: util.HashMap[String, String] = new util.HashMap[String, String]()

    map.put("name", split(1))
    map.put("age", split(2))
    map.put("gender", split(3))
    map.put("clazz", split(4))

    jedis.hmset("student:" + id, map)

  }
}